package com.jumei.flume.sink.kudu;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.jumei.flume.handler.Handler;


/**
 * 
 * @author yihongx
 *
 */
public class KuduSink extends AbstractSink implements Configurable {
    protected Logger logger = LoggerFactory.getLogger(KuduSink.class);

    private final static int DEFAULT_BATCH_SIZE = 1000;
    private int batchSize = 0;

    private List<Event> eventList;

    /**
     */
    Handler handlers;
    
    @Override
    public Status process() throws EventDeliveryException {

        logger.info("sink开始处理");
        Status result = Status.READY;

        Channel channel = getChannel();

        logger.info("渠道名字"+channel.getName());
        Transaction transaction = null;
        
        try {
//            long processedEvents = 0;
            
            transaction = channel.getTransaction();
            transaction.begin();

            eventList.clear();


            for (long processedEvents = 0; processedEvents < batchSize; processedEvents += 1) {
                Event event = channel.take();


                if (event == null) {
                	/**
					 * BUG FIX:
					 *  HIGH CPU LOAD
					 * 
					 */
                    logger.info("无数据");
					if (processedEvents == 0) {
						result = Status.BACKOFF;
						logger.debug("FOUND processedEvents == 0.");
					} else {
						
					}
                    break;
                }

                eventList.add(event);
            }

            try {
				if (CollectionUtils.isNotEmpty(eventList)) {
				    handlers.handler(eventList);
				}
			} catch (Exception ex) {
				logger.error("sensorHandler's error, ", ex);
				throw Throwables.propagate(ex);
			}

            transaction.commit();
        } catch (Exception e) {
        	logger.error("KuduSink's Failed to publish events", e);
            result = Status.BACKOFF;
            try {
            	transaction.rollback();
            } catch (Exception ex) {
                logger.error("Transaction rollback failed", ex);
                throw Throwables.propagate(e);
            }
            throw new EventDeliveryException(e);
        } finally {
        	transaction.close();
        }

        return result;
    }

    @Override
    public void configure(Context context) {
        batchSize = context.getInteger(Config.KUDU_BATCH_SIZE, DEFAULT_BATCH_SIZE);
        eventList = new ArrayList<>(batchSize);
        
        //kudu handler
        String parserClassName = context.getString(Config.KUDU_HANDLER);
        Preconditions.checkArgument(!StringUtils.isEmpty(parserClassName),
                "Must supply a valid handler class name string");
        
        ClassLoader classLoader = getClass().getClassLoader();
        Class parserClass = null;
        try {
            parserClass = classLoader.loadClass(parserClassName);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        Preconditions.checkArgument(parserClass != null,
                "parserClass was misconfigured");
        try {
        	handlers = (Handler) parserClass.newInstance();
            handlers.start();
        } catch (Exception e) {
            logger.error("init KuduSink error!", e);
        }
        
        
    }

    @Override
    public synchronized void start() {
    	if(null == handlers) {
    		handlers.start();
    	}
        super.start();
    }

    @Override
    public synchronized void stop() {
    	if(null != handlers) {
    		handlers.stop();
    	}
    	
        super.stop();
    }

}

